package cn.mesmile.flink.transform;

import cn.mesmile.flink.jdkstream.VideoOrder;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Date;

/**
 * @author zb
 * @date 2021/8/22 23:14
 * @Description
 */
public class FlinkTransformKeyByApp {

    /**
     *  常规的数据流转
     *      DataStream->keyBy操作->KeyStream->window操作->windowStream->聚合操作->DataStream
     *      注意: KeyBy后的聚合函数，只处理当前分组后组内的数据，不同组内数据互不影响
     */

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(2);

        DataStream<VideoOrder> ds = env.fromElements(
                new VideoOrder("21312","java",32,5,new Date()),
                new VideoOrder("314","java",32,5,new Date()),
                new VideoOrder("542","springboot",32,5,new Date()),
                new VideoOrder("42","redis",42,5,new Date()),
                new VideoOrder("52","java",32,5,new Date()),
                new VideoOrder("55","test",52,5,new Date()),
                new VideoOrder("56","cloud",62,5,new Date()),
                new VideoOrder("523","redis",42,5,new Date())
        );

        KeyedStream<VideoOrder, String> videoOrderStringKeyedStream = ds.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder videoOrder) throws Exception {
                return videoOrder.getTitle();
            }
        });

        SingleOutputStreamOperator<VideoOrder> money = videoOrderStringKeyedStream.sum("money");

        money.print("result:");

        env.execute("custom keyBy app");
    }
}
